package cn.itcast.szse;

import cn.itcast.avro.SzseAvro;
import cn.itcast.kafka.KafkaPro;

import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.Socket;
import java.util.Date;

public class SocketClient {
    public static void main(String[] args)  {
        /**
         * 开发步骤:
         * 1.创建main方法
         * 2.建立socket连接，获取流数据
         * 3.解析行数据，数据转换
         * 4.发送kafka
         */
        Socket socket = null;
        try {
            socket = new Socket("localhost",4444);
        InputStream inputStream = socket.getInputStream();

        DataInputStream dataInputStream = new DataInputStream(inputStream);
        KafkaPro kafkaPro=new KafkaPro();

        while (true){
            String str = dataInputStream.readUTF();
            //4.解析行数据，数据转换
            System.out.println(str);
            SzseAvro szseAvro = parseStr(str);
            //5.发送kafka
            kafkaPro.sendDate("szse",szseAvro);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static SzseAvro parseStr(String str) {
        String[] arr = str.split("\\|");
        SzseAvro szseAvro = new SzseAvro();
        szseAvro.setMdStreamID(arr[0].trim());
        szseAvro.setSecurityID(arr[1].trim());
        szseAvro.setSymbol(arr[2].trim());
        szseAvro.setTradeVolume(Long.valueOf(arr[3].trim()));
        szseAvro.setTotalValueTraded(Long.valueOf(arr[4].trim()));
        szseAvro.setPreClosePx(Double.valueOf(arr[5].trim()));
        szseAvro.setOpenPrice(Double.valueOf(arr[6].trim()));
        szseAvro.setHighPrice(Double.valueOf(arr[7].trim()));
        szseAvro.setLowPrice(Double.valueOf(arr[8].trim()));
        szseAvro.setTradePrice(Double.valueOf(arr[9].trim()));
        szseAvro.setClosePx(Double.valueOf(arr[10].trim()));
        szseAvro.setTradingPhaseCode("T01");//实时阶段标志,表示正在交易期间
        szseAvro.setTimestamp(new Date().getTime());
        return szseAvro;

    }
}
